package kafkafile

import (
	"github.com/Shopify/sarama"
	"github.com/sirupsen/logrus"
)

//定义全局变量
var (
	//客户端
	client sarama.SyncProducer
	//接受信息的通道
	MsgChan chan *sarama.ProducerMessage
)

//init是初始化全局的kafka client
func InitKafka(address []string, chanSize int)(err error){
	//生产者配置
	config := sarama.NewConfig()
	//ACK
	config.Producer.RequiredAcks = sarama.WaitForAll
	//分区
	config.Producer.Partitioner = sarama.NewRandomPartitioner
	//确认
	config.Producer.Return.Successes = true

	//连接kafka
	client, err = sarama.NewSyncProducer(address, config)
	if err != nil {
		logrus.Error("kafkafile:producer closed, error:", err)
		return
	}
	//初始化msgChan
	MsgChan = make(chan *sarama.ProducerMessage, chanSize)
	go sendMsg()
	return

}

func sendMsg(){
	for{
		select{
		case msg := <-MsgChan:
			pid, offset, err := client.SendMessage(msg)
			if err != nil {
				logrus.Warning("send message failed,the message is: ", err)
			}
			logrus.Info("send message to kafkafile success. pid%v, offset:%v", pid, offset)
		}
	}
}

// 定义一个函数向外暴露msgChan
func ToMsgChan(msg *sarama.ProducerMessage) {
	MsgChan <-msg
}